package PageRank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class RedTest extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        /**
         * 现在有俩种数据：一种是上轮的投票关系<A,0.5 B   D>
         *                一种是获得的其他页面的权重<A,1.5>  ,<A,0.5>
         */
        try {
            Node oriNode = null;
            double sum = 0.0;
            for (Text v : values) {
                Node node = Node.getNode(v.toString());
                if (node.havingNode()) {
                    //如果node有临近链那么就是第一种
                    oriNode = node;
                } else {
                    //没有就是第二种
                    sum += node.getPr();
                }
            }
            //算出本轮的新pr值
            double newPr = (1 - 0.85) / 4 + 0.85 * sum;
            //算出俩次pr的差值
            double diff = newPr - oriNode.getPr();
            //由于得出的数很小,如果先强转int,会直接变成0
            int diffInt = (int) (diff * 1000);
            //由于计算的是一个波动值,所以这里取绝对值
            diffInt = Math.abs(diffInt);
            System.out.println("本轮与上轮pr的差值" + diffInt);
            //可以求出差值的总和,上面转为int也是为了这里用.这里的increment里面不能放double,只能放整型
            context.getCounter(DriTest.groupName, DriTest.counterName).increment(diffInt);
            //第三件事:写出本轮的投票关系以及新的pr值,<A,1.2 tab B tab D>
            //投票关系没有变,只需要设置一下oriNode对象的pr值为新的pr值即可.这就是map中需要写出上一轮投票关系的原因,因为这里要用
            oriNode.setPr(newPr);
            //一切都是为了计算pr值.考虑到计算pr值需要什么即可
            context.write(key, new Text(oriNode.toString()));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
